[SPARK-24266][K8S][2.4] Restart the watcher when we receive a version changed from k8s by jkleckner · Pull Request #30283 · apache/spark

jkleckner · 2020-11-06T20:49:29Z

What changes were proposed in this pull request?

This patch processes the HTTP Gone event and restarts the pod watcher.

Why are the changes needed?

This is a backport of PR #28423 to branch-2.4.
The reasons are explained in SPARK-24266 that spark jobs using the k8s resource scheduler may hang.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Manually.

jkleckner · 2020-11-06T20:51:05Z

@shockdm Please check this over.

@dongjoon-hyun Maybe we can get this into 2.4 ?

My force-push prevented the old PR from being reopened it seems.

dongjoon-hyun · 2020-11-06T22:02:36Z

ok to test

SparkQA · 2020-11-06T22:23:51Z

Test build #130737 has finished for PR 30283 at commit 404bfa4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-11-06T22:26:12Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35346/

SparkQA · 2020-11-06T22:37:59Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35346/

jkleckner · 2020-11-06T23:04:32Z

An image of just the spark build was created with this job [1].

We are using this image starting today in our dev cluster.

[1] https://gitlab.com/jkleckner/spark/-/pipelines/212962140

jkleckner · 2020-11-09T19:58:01Z

This version of Spark has been running fine since the last 3 days in our cluster.

dongjoon-hyun · 2020-11-09T23:49:13Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

 private[k8s] trait LoggingPodStatusWatcher extends Watcher[Pod] {
-  def awaitCompletion(): Unit
+  def reset(): Unit
+  def watchOrStop(sId: String): Boolean


Could you preserve the order from the original patches (master/branch-3.0)?

def watchOrStop(submissionId: String): Boolean def reset(): Unit

The function declaration order is different. The parameter name is not the same here (sId != submissionId)

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

dongjoon-hyun · 2020-11-09T23:52:34Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

  override def onClose(e: KubernetesClientException): Unit = {
-    logDebug(s"Stopping watching application $appId with last-observed phase $phase")
-    closeWatch()
+    logInfo(s"Stopping watching application $appId with last-observed phase $phase")


This is a regression because this is previous logDebug. (Of course, this is different from master branch, too).

dongjoon-hyun · 2020-11-09T23:52:57Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

+    logInfo(s"Stopping watching application $appId with last-observed phase $phase")
+    if (e != null && e.getCode==HTTP_GONE) {
+      resourceTooOldReceived = true
+      logInfo(s"Got HTTP Gone code, resource version changed in k8s api: $e")


dongjoon-hyun · 2020-11-09T23:53:06Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

+      resourceTooOldReceived = true
+      logInfo(s"Got HTTP Gone code, resource version changed in k8s api: $e")
+    } else {
+      logInfo(s"Got proper termination code, closing watcher.")


Please remove this.

dongjoon-hyun · 2020-11-09T23:54:23Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

-      s"Container final statuses:\n\n${containersDescription(p)}"
-    }.getOrElse("No containers were found in the driver pod."))
-  }
-


This removal looks like a part of independency PR instead of the part of SPARK-24266. Could you tell us why this is required and where this came from?

@shockdm Could you chime in on this one?

@dongjoon-hyun It does look like it originates from SPARK-28947 02c5b4f which eliminated the future and was a rename.

Since this is a private trait, the logic should be completely self-contained and safe to remove.

@jkleckner thank you for following up, that is correct. Sorry for the late response :(

dongjoon-hyun · 2020-11-09T23:54:53Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

  }
+
+  override def watchOrStop(sId: String): Boolean = if (waitForCompletion) {
+    logInfo(s"Patched Sept 8th: Waiting for application" +


Patched Sept 8th:?

dongjoon-hyun · 2020-11-09T23:55:52Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

+    synchronized {
+      while (!podCompleted && !resourceTooOldReceived) {
+        wait(interval.get)
+        logDebug(s"Application status for $appId (phase: $phase)")


This should be logInfo.

dongjoon-hyun · 2020-11-09T23:58:43Z

...managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala

    createdResourcesArgumentCaptor = ArgumentCaptor.forClass(classOf[HasMetadata])
    when(podOperations.create(FULL_EXPECTED_POD)).thenReturn(POD_WITH_OWNER_REFERENCE)
    when(namedPods.watch(loggingPodStatusWatcher)).thenReturn(mock[Watch])
+    when(loggingPodStatusWatcher.watchOrStop("default" + ":" + POD_NAME)).thenReturn(true)


Maybe, kubernetesConf.namespace() instead of "default"?

dongjoon-hyun

Hi, @jkleckner .
This will be an official Apache Spark patch. Could you clean up something like logInfo(s"Patched Sept 8th: and make it consistent across Apache Spark branches?

… changed from k8s This is a backport of apache#29533 from master. It includes the shockdm/pull/1 which has been squashed and the import review comment include. It has also been rebased to branch-2.4 Address review comments.

jkleckner · 2020-11-10T02:39:26Z

@dongjoon-hyun Sorry for doing force-push. I got used to gitlab which keeps track of history better.

SparkQA · 2020-11-10T02:57:35Z

Test build #130823 has finished for PR 30283 at commit 5960d24.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-11-10T03:30:09Z

Test build #130825 has finished for PR 30283 at commit 86f8ee8.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-11-11T16:53:31Z

Retest this please

SparkQA · 2020-11-11T17:06:08Z

Test build #130946 has finished for PR 30283 at commit 86f8ee8.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-11-15T03:59:44Z

Retest this please.

dongjoon-hyun · 2020-11-15T04:02:51Z

...tes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala

+        val podWithName = kubernetesClient
+          .pods()
+          .withName(driverPodName)
+


Could you add the following comments like the other branches?

// Reset resource to old before we start the watch, this is important for race conditions

dongjoon-hyun · 2020-11-15T04:03:16Z

...tes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala

+          .withName(driverPodName)
+
+        watcher.reset()
+


Let's remove this empty line like the other branch.

dongjoon-hyun · 2020-11-15T04:13:11Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

+    podCompleted
+  } else {
+    logInfo(s"Deployed Spark application ${appId} with submission ID $sId into Kubernetes")
+    logInfo(s"It seems we end up here, because we never want to wait for completion...")


Please remote this.

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

dongjoon-hyun · 2020-11-15T04:15:02Z

...managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala

      KUBERNETES_RESOURCE_PREFIX)
    submissionClient.run()
-    verify(loggingPodStatusWatcher).awaitCompletion()
+    verify(loggingPodStatusWatcher).watchOrStop("default:driver")


default -> kubernetesConf.namespace()?

dongjoon-hyun

Thank you for updating, @jkleckner . I left some comments to match this to the other branches.

SparkQA · 2020-11-15T04:16:48Z

Test build #131105 has finished for PR 30283 at commit 86f8ee8.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-11-15T04:20:52Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

  }
+
+  override def watchOrStop(sId: String): Boolean = if (waitForCompletion) {
+    logInfo(s"Waiting for application ${conf.appName} with submission ID $sId to finish...")


Oh, it's a compilation error.

[error] /home/jenkins/workspace/SparkPullRequestBuilder@2/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala:195: object appName is not a member of package conf [error] logInfo(s"Waiting for application ${conf.appName} with submission ID $sId to finish...") [error] ^ [error] one error found

Can you confirm that this is the way you want it?

dongjoon-hyun · 2020-11-15T04:21:30Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

  override def onClose(e: KubernetesClientException): Unit = {
    logDebug(s"Stopping watching application $appId with last-observed phase $phase")
-    closeWatch()
+    if (e != null && e.getCode==HTTP_GONE) {


e.getCode == HTTP_GONE?

Missed this one somehow. Updated now.

dongjoon-hyun

For LoggingPodStatusWatcherImpl, I missed that it doesn't have conf which is different from the other branch class LoggingPodStatusWatcherImpl(conf: KubernetesDriverConf). Sorry about that. Could you fix the compilation error?

jkleckner · 2020-11-20T01:08:54Z

@dongjoon-hyun I made the changes. No force-push this time.

Sorry to take so long but I did get very very busy.

SparkQA · 2020-11-20T01:35:53Z

Test build #131385 has finished for PR 30283 at commit f01b3d3.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-11-20T02:12:36Z

Test build #131388 has finished for PR 30283 at commit 1c64c6c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-11-20T02:26:36Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35988/

SparkQA · 2020-11-20T02:39:02Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35988/

SparkQA · 2020-11-20T03:02:17Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35991/

SparkQA · 2020-11-20T03:15:52Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35992/

SparkQA · 2020-11-20T03:22:46Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35991/

SparkQA · 2020-11-20T03:28:01Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35992/

dongjoon-hyun · 2020-11-24T18:16:56Z

...tes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala

-    val watcher = new LoggingPodStatusWatcherImpl(kubernetesAppId, loggingInterval)
+    val watcher = new LoggingPodStatusWatcherImpl(kubernetesAppId,
+      loggingInterval,
+      waitForAppCompletion)


~~Please revert this change. This is inconsistent with Apache Spark 3.1 and 3.0.~~

dongjoon-hyun · 2020-11-24T18:17:02Z

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala

    appId: String,
-    maybeLoggingInterval: Option[Long])
+    maybeLoggingInterval: Option[Long],
+    waitForCompletion: Boolean)


~~Please revert this change. This is inconsistent with Apache Spark 3.1 and 3.0.~~

… changed from k8s ### What changes were proposed in this pull request? This patch processes the HTTP Gone event and restarts the pod watcher. ### Why are the changes needed? This is a backport of PR #28423 to branch-2.4. The reasons are explained in SPARK-24266 that spark jobs using the k8s resource scheduler may hang. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Manually. Closes #30283 from jkleckner/shockdm-2.4.6-spark-submit-fix. Lead-authored-by: Jim Kleckner <jim@cloudphysics.com> Co-authored-by: Dmitriy Drinfeld <dmitriy.drinfeld@ibm.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

dongjoon-hyun

Thank you, @jkleckner and all.
+1, LGTM. Merged to branch-2.4 for Apache Spark 2.4.8.

jkleckner · 2020-11-24T19:06:50Z

Thank you, @dongjoon-hyun !

jkleckner changed the title ~~[SPARK-24266][K8S][2.4] Restart the watcher when we receive a version…~~ [SPARK-24266][K8S][2.4] Restart the watcher when we receive a version changed from k8s Nov 6, 2020

dongjoon-hyun reviewed Nov 9, 2020

View reviewed changes

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala Show resolved Hide resolved

dongjoon-hyun reviewed Nov 9, 2020

View reviewed changes

dongjoon-hyun requested changes Nov 10, 2020

View reviewed changes

jkleckner force-pushed the shockdm-2.4.6-spark-submit-fix branch from 404bfa4 to 5960d24 Compare November 10, 2020 02:32

jkleckner force-pushed the shockdm-2.4.6-spark-submit-fix branch from 5960d24 to 86f8ee8 Compare November 10, 2020 02:36

dongjoon-hyun added the KUBERNETES label Nov 15, 2020

dongjoon-hyun reviewed Nov 15, 2020

View reviewed changes

...ernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/LoggingPodStatusWatcher.scala Show resolved Hide resolved

dongjoon-hyun reviewed Nov 15, 2020

View reviewed changes

Address review comments

f01b3d3

One more review change

1c64c6c

jkleckner force-pushed the shockdm-2.4.6-spark-submit-fix branch from b4cc870 to 1c64c6c Compare November 20, 2020 01:49

jkleckner requested a review from dongjoon-hyun November 20, 2020 16:20

dongjoon-hyun reviewed Nov 24, 2020

View reviewed changes

dongjoon-hyun approved these changes Nov 24, 2020

View reviewed changes

dongjoon-hyun closed this Nov 24, 2020

jkleckner deleted the shockdm-2.4.6-spark-submit-fix branch November 24, 2020 19:06

jkleckner mentioned this pull request Feb 16, 2021

Running different version of spark jobs on a spark-operator cluster kubeflow/spark-operator#1163

Closed

jkleckner mentioned this pull request Oct 16, 2021

Spark batch jobs are not terminating kubeflow/spark-operator#627

Closed

Conversation

jkleckner commented Nov 6, 2020 • edited by dongjoon-hyun Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

jkleckner commented Nov 6, 2020

Uh oh!

dongjoon-hyun commented Nov 6, 2020

Uh oh!

SparkQA commented Nov 6, 2020

Uh oh!

SparkQA commented Nov 6, 2020

Uh oh!

SparkQA commented Nov 6, 2020

Uh oh!

jkleckner commented Nov 6, 2020

Uh oh!

jkleckner commented Nov 9, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jkleckner commented Nov 10, 2020

Uh oh!

SparkQA commented Nov 10, 2020

Uh oh!

SparkQA commented Nov 10, 2020

Uh oh!

dongjoon-hyun commented Nov 11, 2020

Uh oh!

SparkQA commented Nov 11, 2020

Uh oh!

dongjoon-hyun commented Nov 15, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Nov 15, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

jkleckner commented Nov 6, 2020 •

edited by dongjoon-hyun

Loading

jkleckner commented Nov 9, 2020 •

edited

Loading

dongjoon-hyun left a comment •

edited

Loading

dongjoon-hyun Nov 15, 2020 •

edited

Loading